package com.lizhiyu.flink.demo1;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


/***
 * Flink 流处理   因为使用的DataStream
 *
 * 监听socket获得数据源，然后将数据打印出来
 * 创建时通可以有ui界面进行访问
 */
public class A1_Flink_UI {
    public static void main(String[] args) throws Exception {
        //下载  netcat
        //(下载成功后)
        //windows 执行 nc -l -p 8888

        //linux 安装 yum install -y netcat
        //          yum install -y nc
        //linux 执行 nc -lk 8888
        //启动项目在cmd中发送信息即可接收
        //在 localhost:8081 中即可看到正在工作的JOB
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //env.setParallelism(1);
        DataStream<String> stringDataStream = env.socketTextStream("127.0.0.1",8888);
        DataStream<String> flatMapDataStream = stringDataStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] arr = value.split(",");
                for (String word : arr) {
                    out.collect(word);
                }
            }
        });
        flatMapDataStream.print();
        //DataStream需要调用execute,可以取个名称
        env.execute("data stream job");
    }
}
